package com.liuqi.iot.collector;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.liuqi.common.utils.OgnlUtils;
import com.liuqi.iot.bean.ScriptInfo;
import com.liuqi.iot.bean.dto.GatewayDTO;
import com.liuqi.iot.collector.parser.OriginDataParser;
import com.liuqi.iot.collector.processor.NoKafkaDataProcessor;
import com.liuqi.utils.GroovyUtils;
import com.liuqi.utils.JsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;

/**
 * 原始工况处理器
 * 接收到数据后不做处理直接传到kafka原始工况主题中
 *
 * @author LiuQi 2025/3/13-9:21
 * @version V1.0
 **/
@Slf4j
public abstract class AbstractDataCollector {
    @Autowired(required = false)
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${iot.kafka.topic.origin}")
    private String originTopic;

    @Autowired(required = false)
    private NoKafkaDataProcessor noKafkaDataProcessor;

    @Autowired
    private OriginDataParser dataParser;

    // 停止监听
    public abstract void stopListener(String gatewayId);

    /**
     * 启动MQTT监听
     * @param gateway 网关
     */
    public abstract void startListener(GatewayDTO gateway) throws Exception;

    /**
     * 处理消息内容
     *
     * @param msg      接收到的消息
     */
    protected void process(GatewayDTO gateway, String topic, String msg) {
        if (StringUtils.isBlank(msg)) {
            return;
        }

        Object obj = JSON.parse(msg);

        // 数据预处理
        ScriptInfo scriptInfo = gateway.getConverter();
        if (null != scriptInfo && StringUtils.isNotBlank(scriptInfo.getScript())) {
            String scriptType = scriptInfo.getType();
            String script = scriptInfo.getScript();
            if ("js".equals(scriptType) || "javascript".equals(scriptType)) {
                obj = JsUtils.execute(script, obj);
            } else if ("groovy".equals(scriptType)){
                obj = GroovyUtils.execute(script, obj);
            } else if ("ognl".equals(scriptType)) {
                obj = OgnlUtils.execute(script, obj);
            }

            if (null == obj || "null".equals(obj) || "false".equals(obj)) {
                // 不需要处理的消息
                return;
            }

            obj = JSON.parse(JSON.toJSONString(obj));
        }

        String deviceCode = gateway.parseDeviceCode(topic);
        Map<String, List<JSONObject>> workDataMap = dataParser.parse(gateway, deviceCode, obj);

        // 如果启用了kafka，则将数据推送原始工况队列
        if (null != kafkaTemplate) {
            workDataMap.forEach((realDeviceCode, list) -> {
                list.forEach(map -> kafkaTemplate.send(originTopic, gateway.getTenantId(), JSON.toJSONString(map)));
            });
        } else {
            // 否则，直接处理数据
            if (null != noKafkaDataProcessor) {
                workDataMap.forEach((realDeviceCode, list) -> noKafkaDataProcessor.process(realDeviceCode, list));
            }
        }
    }

    /**
     * 解析上云时间
     */
    private Long parseTimestamp(GatewayDTO gateway, JSONObject obj) {
        String tsField = MapUtils.getString(gateway.getMetadata(), "tsField", "timestamp");
        Object timestampObj = obj.get(tsField);
        if (null == timestampObj) {
            return LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
        }

        if (timestampObj instanceof Long l) {
            return l;
        } else {
            // 字符串
            String str = timestampObj.toString();
            DateTimeFormatter formatter;
            if (str.contains(".")) {
                formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
            } else {
                formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            }

            return LocalDateTime.parse(timestampObj.toString(), formatter).toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
        }
    }
}
